package com.mango.ch06;

import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import com.mango.Tools.DateUtils;

public class AverageReducer extends Reducer<ComposeKey, NaturalValue, Text, Text> {
	private int winSize = 3;

	@Override
	protected void reduce(ComposeKey key, Iterable<NaturalValue> values, Context context)
			throws IOException, InterruptedException {
		SimpleMovingAverage sma = new SimpleMovingAverage(winSize);
		for (NaturalValue value : values) {
			sma.addNewNumber(value.getValue());
			double result = sma.getMovingAverage();
			String str = DateUtils.localDateFormat(value.getTimeStamp()) + "," + result;
			context.write(new Text(key.getName()), new Text(str));
		}
		// TODO Auto-generated method stub

	}

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		this.winSize = Integer.valueOf(context.getConfiguration().get("winSize"));
	}

}
